
package lazyPB;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import probabilisticBroadcast.PbBroadcast;
import probabilisticBroadcast.PbDeliver;
import probabilisticBroadcast.ProbabilisticBroadcast;
import se.kth.ict.id2203.flp2p.FairLossPointToPointLink;
import se.kth.ict.id2203.flp2p.Flp2pDeliver;
import se.kth.ict.id2203.flp2p.Flp2pSend;
import se.sics.kompics.ComponentDefinition;
import se.sics.kompics.Handler;
import se.sics.kompics.Kompics;
import se.sics.kompics.Negative;
import se.sics.kompics.Positive;
import se.sics.kompics.Start;
import se.sics.kompics.address.Address;
import se.sics.kompics.timer.ScheduleTimeout;
import se.sics.kompics.timer.Timer;
import unrealiableBroadcast.UnBroadcast;
import unrealiableBroadcast.UnDeliver;
import unrealiableBroadcast.UnreliableBroadcast;

/**
 * The <code>SimpleUnreliableBroadcast</code> class.
 * 
 * @author Federico Piccinini <fpic@sics.se>
 */
public final class LazyPB extends ComponentDefinition {

	Positive<FairLossPointToPointLink> flp2p = requires (FairLossPointToPointLink.class);
	Positive<UnreliableBroadcast> un =  requires (UnreliableBroadcast.class);	
	Positive<Timer> timer = requires (Timer.class);
	
	Negative<ProbabilisticBroadcast> pb = provides(ProbabilisticBroadcast.class);

	private static final Logger logger = LoggerFactory
			.getLogger(LazyPB.class);

	private Set<Address> neighborSet;
	private Address self; 
	private Map<Address, Integer> delivered; /* Sequence number of last message delivered sent by process p */
	private int lsn; /* lsn = local sequence number */
	private Set<DataMessage> pending;
	private Set<DataMessage> stored;
	private int fanout;
	private double storeThreshold;
	private int maxRounds;
	private int timeDelay;
	

	/**
	 * Instantiates a new application0.
	 */
	public LazyPB() {
		subscribe(handleInit, control);
		subscribe(handleStart, control);
		subscribe(handlePbBroadcast, pb);
		subscribe(handleUnDeliver, un);
		subscribe(requestMessageHandler, flp2p);
		subscribe(dataMessageHandler, flp2p);
		subscribe(timeoutHandler, timer);
	}

	Handler<LazyPBInit> handleInit = new Handler<LazyPBInit>() {
		public void handle(LazyPBInit event) {
			logger.info("Received Init event");
			neighborSet = event.getNeighborSet();
			self = event.getSelf();
			delivered = new HashMap<Address, Integer>();
			for (Address neighbor: neighborSet) {
				delivered.put(neighbor, 0);
			}
			lsn = 0;
			pending = new HashSet<DataMessage>();
			stored = new HashSet<DataMessage>();
			fanout = event.getFanout();
			storeThreshold = event.getStoreThreshold();
			maxRounds = event.getMaxRounds();
			timeDelay = event.getTimeDelay();
		}
	};

	Handler<Start> handleStart = new Handler<Start>() {
		public void handle(Start event) {
		}
	};
	
	private void deliverPending (Address s) {
		HashSet<DataMessage> newPending = new HashSet<DataMessage>();
		boolean completed = false;
		
		while (!completed) {
			completed = true;
			for (DataMessage msg: pending) {
				DataMessage dataMsg = (DataMessage) msg;
				if (dataMsg.getSequence() == delivered.get(s)+1
						&& dataMsg.getSource().equals(s)) {
					delivered.put(s, dataMsg.getSequence());
					completed = false;
					//this is the deliver event:
					trigger (dataMsg, pb);
				} else {
					newPending.add(dataMsg);
				}
			}
			pending = newPending;
		}
		
	}
	
	private void gossip (PbDeliver msg) {
		Set<Address> targets = pickTargets (fanout);
		if (targets==null) return;
		for (Address t : targets) {
			trigger( new Flp2pSend(t, msg), flp2p);
		}
	}
	
	private Set<Address> pickTargets (int nTargets) {
		if (nTargets > neighborSet.size())
			return null;
		Set<Address> targets = new HashSet<Address>();
		List<Address> neighborList = new ArrayList<Address> (neighborSet);
		Random randomGenerator = new Random ();
		while (targets.size() < nTargets) {
			int rnd = randomGenerator.nextInt(neighborList.size());
			Address candidate = neighborList.get(rnd);
			if (!targets.contains(candidate) && candidate!=self) {
				targets.add(candidate);
			}
		}
		return targets;
	}
	
	Handler<PbBroadcast> handlePbBroadcast = new Handler<PbBroadcast> () {
		public void handle(PbBroadcast event) {
			if (event.getDeliverEvent() instanceof DataMessage) {
				lsn++;
				DataMessage dataMsg = (DataMessage) event.getDeliverEvent();
				logger.info("Broadcasting message {} with sn {}", dataMsg.getMessage(), lsn);
				trigger (new UnBroadcast(new UnDeliver(self, 
						new DataMessage(self, dataMsg.getMessage(), lsn))), un);
			}
		}
	};

	Handler <UnDeliver> handleUnDeliver = new Handler<UnDeliver> () {
		public void handle (UnDeliver event) {
			if (event.getMessage() instanceof DataMessage) {
				DataMessage dataMsg = (DataMessage) event.getMessage();
				Random randomGen = new Random();
				Address src = dataMsg.getSource();
				if (dataMsg.isGossip()) return;
				
				if (randomGen.nextFloat()>storeThreshold) {
					stored.add(dataMsg);
				}

				if (dataMsg.getSequence() == (delivered.get(src)+1)) {
					delivered.put(src, dataMsg.getSequence());
					logger.info("Correctly received message with sequence number {} from process {}", 
							dataMsg.getSequence(), dataMsg.getSource());
					//this is the deliver event:
					trigger (dataMsg, pb);
				} else if (dataMsg.getSequence() > (delivered.get(src)+1)) {
					logger.info("Wrongly received message with sequence number {} from process{}", 
							dataMsg.getSequence(), dataMsg.getSource());
					pending.add(dataMsg);
					for (int seqNum = delivered.get(src)+1; seqNum<dataMsg.getSequence(); ++seqNum) {
						logger.info("Trying to retrieve message with sequence number {} from process {}", seqNum, src);
						gossip (new RequestMessage (self, dataMsg.getSource(), seqNum, maxRounds-1));
					}
					startTimer(src, dataMsg.getSequence());
				}	
			}
		}
	};
	
	private void startTimer (Address p, int seqNo) {
		//logger.info("Scheduling check timeout in {} milliseconds...", delay);
		ScheduleTimeout st = new ScheduleTimeout(timeDelay);
		st.setTimeoutEvent(new LPBTimeout(st, p, seqNo));
		trigger(st, timer);
	}
	
	Handler <RequestMessage> requestMessageHandler = new Handler <RequestMessage> () {
		public void handle(RequestMessage event) {
			DataMessage storedMsg = isStored(event.getMsgSrc(), event.getSeqNum(), stored);
			if (storedMsg != null) {
				DataMessage msg = new DataMessage(storedMsg.getSource(), 
						storedMsg.getMessage(), storedMsg.getSequence(), true);
				trigger (new Flp2pSend(event.getSource(), msg), flp2p);
			} else if (event.getRounds()>0) {
				gossip (new RequestMessage(event.getSource(), event.getMsgSrc(), 
						event.getSeqNum(), event.getRounds()-1));
			}
			
		}
	};
	
	Handler <DataMessage> dataMessageHandler = new Handler<DataMessage> () {
		@Override
		public void handle(DataMessage event) {
			if (event.isGossip()) {
				logger.info("Retrieved message {} from process {} via gossiping.",
						event.getSequence(), event.getSource());
				if (event.getSequence() == delivered.get(event.getSource())+1) {
					Address src = event.getSource();
					int seqNo = event.getSequence();
					delivered.put(src, seqNo);
					//this is the deliver event:
					trigger (event, pb);
					deliverPending(src);
				} else {
					pending.add(event);
				}
			}
		}
		
	};
	
	
	DataMessage isStored (Address src, int seqNo, Set<DataMessage> msgSet) {
		for (DataMessage msg: msgSet) {
			if (msg.getSource().equals(src) && msg.getSequence()==seqNo) {
				return msg;
			}
		}
		return null;
	}
	
	Handler<LPBTimeout> timeoutHandler = new Handler<LPBTimeout>() {
		@Override
		public void handle(LPBTimeout event) {
			logger.info("Received timeout event for message {} from process {}",
					event.getSeqNo(), event.getProcess());
			/*if (event.getSeqNo() == delivered.get(event.getProcess())+1) {
				delivered.put(event.getProcess(), event.getSeqNo());
				logger.info("Counter for process {} updated to {}", event.getProcess(), event.getSeqNo());
			}*/
			for (int seqno = delivered.get(event.getProcess())+1; seqno <= event.getSeqNo(); ++seqno) {
				DataMessage dataMsg = isStored (event.getProcess(), seqno, pending);
				if (dataMsg!= null) {
					trigger (dataMsg, pb);
				}
				pending.remove(dataMsg);
			}
			delivered.put(event.getProcess(), event.seqNo);
			
		}
		
	};
	
	private void doShutdown() {
		System.out.println("2DIE");
		System.out.close();
		System.err.close();
		Kompics.shutdown();
	}
}
